package com.study.flink.example.loginfail

import org.apache.flink.api.scala._
import org.apache.flink.cep.scala.CEP
import org.apache.flink.cep.scala.pattern.Pattern
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.api.windowing.time.Time

/**
  * 登录异常检测
  * 在2秒内连续两次登录失败发发出警报
  *
  * @author stephen
  * @date 2019-07-24 20:46
  */
// 输入的登录事件流
case class LoginEvent(userId: Int, ip: String, eventType: String, eventTime: Long)

// 输出的报警信息
case class Warning(userId: Long, firstFailTime: Long, lastFailTime: Long, warningMsg: String)

object LoginFailDetect {

  def main(args: Array[String]): Unit = {

    // 1 获取环境
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    env.setParallelism(1)

    // 2 获取数据

    val loginStream = env.fromCollection(List(
      LoginEvent(1, "192.168.0.1", "fail", 1558430842),
      LoginEvent(1, "192.168.0.2", "fail", 1558430843),
      LoginEvent(1, "192.168.0.3", "fail", 1558430844),
      LoginEvent(2, "192.168.10.10", "success", 1558430845),
      LoginEvent(2, "192.168.10.10", "fail", 1558430846),
      LoginEvent(2, "192.168.10.10", "fail", 1558430847),
      LoginEvent(2, "192.168.10.10", "fail", 1558430848)
    ))
      .assignAscendingTimestamps(_.eventTime * 1000)

    // 定义pattern，对事件流进行模式匹配
    val loginFailPattern = Pattern.begin[LoginEvent]("first")
      .where(_.eventType == "fail")
      .next("second")
      .where(_.eventType == "fail")
      .within(Time.seconds(
        3))

    // 在输入流的基础上应用pattern，得到匹配的pattern stream
    val patternStream = CEP.pattern(loginStream.keyBy(_.userId), loginFailPattern)

    import scala.collection.Map
    val loginFailStream = patternStream.select((patternEvents: Map[String, Iterable[LoginEvent]]) => {
      val firstFailEvent = patternEvents.getOrElse("first", null).iterator.next()
      val secondFailEvent = patternEvents.getOrElse("second", null).iterator.next()
      Warning(firstFailEvent.userId, firstFailEvent.eventTime, secondFailEvent.eventTime, "login fail waring")
    })

    // 4 输出结果
    loginFailStream.print("异常登录")

    // 5 启动任务
    env.execute("Login Fail Detect with CEP")
  }
}
